package org.lagou.edu.stock.task;

import com.alibaba.fastjson2.JSON;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer;
import org.lagou.edu.stock.model.Stock;
import org.lagou.edu.stock.utils.RedisUtil;

/**
 * @Author 咖啡杯里的茶
 * @date 2022/7/16
 */
public class ListenBinlogTask implements Runnable{
	@Override
	public void run() {
		BinaryLogClient client = new BinaryLogClient("localhost", 3306, "root", "admin");
		EventDeserializer eventDeserializer = new EventDeserializer();
		client.setEventDeserializer(eventDeserializer);
		client.registerEventListener(new BinaryLogClient.EventListener() {
			@Override
			public void onEvent(Event event) {
				if(!matchTable(event.getData())) {
					return;
				}
				String rowData;
				EventHeader header = event.getHeader();
				EventType eventType = header.getEventType();
				System.out.println("监听的事件类型:" + eventType);
				if (EventType.isWrite(eventType)) {
					WriteRowsEventData data = event.getData();
					rowData = JSON.toJSONString(data);
				} else if (EventType.isUpdate(eventType)) {
					UpdateRowsEventData data = event.getData();
					rowData = JSON.toJSONString(data);
				} else {
					return;
				}
				Stock stock = JSON.parseObject(rowData, Stock.class);
				RedisUtil.syncStock(stock.getShopId(), stock.getQuantity());
			}
		});
		//client.connect();
	}

	/**
	 * 第一类监听事件：判断是否是库存表发生了改变
	 * @param data
	 * @return
	 */
	private boolean matchTable(EventData data) {
		if (data instanceof TableMapEventData) {
			TableMapEventData tableMapEventData = (TableMapEventData) data;
			// mysql 中库存表名称就是 stock
			return "stock".equals(tableMapEventData.getTable());
		}
		return false;
	}
}
